package com.bins.cloud.stream.mq;

import com.bins.cloud.stream.eo.mq.MqMsg;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.common.message.MessageConst;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.ApplicationRunner;
import org.springframework.cloud.stream.function.StreamBridge;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.GenericMessage;

import java.util.HashMap;
import java.util.Map;

@Slf4j
@Configuration
public class StreamOrderlyProducer {

    public static final String[] tags = new String[]{"TagA", "TagB", "TagC", "TagD", "TagE"};

    @Autowired
    private StreamBridge streamBridge;

    @Bean
    public ApplicationRunner orderlyMqProducer() {
        return args -> {
            for (int i = 0; i < 100; i++) {
                String key = "KEY" + i;
                Map<String, Object> headers = new HashMap<>();
                headers.put(MessageConst.PROPERTY_KEYS, key);
                headers.put(MessageConst.PROPERTY_TAGS, tags[i % tags.length]);
                headers.put(MessageConst.PROPERTY_ORIGIN_MESSAGE_ID, i);
                MqMsg mqMsg = new MqMsg();
                mqMsg.setId(String.valueOf(i));
                mqMsg.setName(tags[i % tags.length]);
                Message<MqMsg> msg = new GenericMessage(mqMsg, headers);
                streamBridge.send("orderlyProducer-out-0", msg);
            }
        };
    }
}